package cn.com.scooper.impl.ks;

import cn.com.scooper.common.constant.CommonConstant;
import cn.com.scooper.common.util.StringUtils;
import cn.com.scooper.dao.xf.MineIotDeviceDataDao;
import cn.com.scooper.pojo.dto.ks.ReportRequestPkgDto;
import cn.com.scooper.pojo.dto.ks.ReportResponsePkgDto;
import cn.com.scooper.pojo.po.MineIotDeviceDataPo;
import cn.com.scooper.service.ks.MessageCallback;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import cn.com.scooper.impl.netty.ReportNettyClient;

@Service
@Slf4j
public class KSReportDataHandler implements MessageCallback {

    private ReportNettyClient reportNettyClient;

    private ConcurrentHashMap<String, LinkedBlockingQueue<ReportRequestPkgDto>> concurrentHashMap;

    // 是否可以继续发送下一个报文的信号量
    private Semaphore semaphoreHSSN;
    private Semaphore semaphoreNFSN;
    private Semaphore semaphoreCYKY;

    // 最新一个发送的报文
    private ReportRequestPkgDto reportPkgHSSN;
    private ReportRequestPkgDto reportPkgNFSN;
    private ReportRequestPkgDto reportPkgCYKY;

    // 报文未发送成功，重新发送次数
    private AtomicInteger reSendNumHSSN;
    private AtomicInteger reSendNumNFSN;
    private AtomicInteger reSendNumCYKY;

    // 一直没收到响应消息
    private ScheduledExecutorService executorServiceHSSN;
    private Future<?> timeoutTaskHSSN;
    private ScheduledExecutorService executorServiceNFSN;
    private Future<?> timeoutTaskNFSN;
    private ScheduledExecutorService executorServiceCYKY;
    private Future<?> timeoutTaskCYKY;

    @Autowired
    private MineIotDeviceDataDao mineIotDeviceDataDao;

    public KSReportDataHandler(ReportNettyClient reportNettyClient){
        this.reportNettyClient = reportNettyClient;
        this.reportNettyClient.setKsReportDataHandler(this);

        concurrentHashMap = new ConcurrentHashMap<>();
        concurrentHashMap.put(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN, new LinkedBlockingQueue<>(1000));
        concurrentHashMap.put(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN, new LinkedBlockingQueue<>(1000));
        concurrentHashMap.put(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY, new LinkedBlockingQueue<>(1000));

        semaphoreHSSN = new Semaphore(1);
        semaphoreNFSN = new Semaphore(1);
        semaphoreCYKY = new Semaphore(1);

        executorServiceHSSN = Executors.newSingleThreadScheduledExecutor();
        executorServiceNFSN = Executors.newSingleThreadScheduledExecutor();
        executorServiceCYKY = Executors.newSingleThreadScheduledExecutor();

        new Thread(new NFSNDataConsumeThread(),"NFSNDataConsumeThread").start();
        new Thread(new HSSNDataConsumeThread(),"HSSNDataConsumeThread").start();
        new Thread(new CYKYDataConsumeThread(),"CYKYDataConsumeThread").start();
    }

    public boolean offer(ReportRequestPkgDto reportRequestPkgDto, String enterpriseName){
        if( !reportNettyClient.isConnected() ){
            log.info("【企业实时数据接收】上报企业：{}，市局上报netty连接不可用，报文存入数据库",enterpriseName);
            MineIotDeviceDataPo mineIotDeviceDataPo = new MineIotDeviceDataPo();
            mineIotDeviceDataPo.setReportData(JSONObject.toJSONString(reportRequestPkgDto));
            mineIotDeviceDataPo.setIsReported((byte)0);
            mineIotDeviceDataPo.setReportEnterprise(enterpriseName);
            mineIotDeviceDataPo.setGmtCreate(new Date());
            mineIotDeviceDataPo.setGmtUpdate(new Date());
            return mineIotDeviceDataDao.insert(mineIotDeviceDataPo)>0;
        }
        return concurrentHashMap.get(enterpriseName).offer(reportRequestPkgDto);
    }

    public ReportRequestPkgDto poll(String enterpriseName){
        return concurrentHashMap.get(enterpriseName).poll();
    }

    /**
     * 消费数据线程
     */
    private class NFSNDataConsumeThread implements Runnable {
        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    if( reportNettyClient.isConnected() ){
                        // 在收到前一个报文接收成功的响应信息后才可传递下一个报文
                        semaphoreNFSN.acquire();
                        // 从队列中取出数据，如果队列为空则阻塞等待
                        ReportRequestPkgDto data = concurrentHashMap.get(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN).take();
                        log.info("【向市局上报数据->从队列取出南方水泥数据发送】报文dataId：{}",data.getDataId());
                        // 消费数据的逻辑
                        sendReportPackage(data);
                        reportPkgNFSN = data;
                        reSendNumNFSN = new AtomicInteger(3);
                        createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN);
                    }else{
                        Thread.sleep(30_000);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    semaphoreNFSN.release();
                    log.error("【向市局上报数据->从队列取出南方水泥数据发送】处理中断：{}",e.getMessage());
                }
            }
        }
    }

    private class HSSNDataConsumeThread implements Runnable {
        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    if( reportNettyClient.isConnected() ){
                        // 在收到前一个报文接收成功的响应信息后才可传递下一个报文
                        semaphoreHSSN.acquire();
                        // 从队列中取出数据，如果队列为空则阻塞等待
                        ReportRequestPkgDto data = concurrentHashMap.get(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN).take();
                        log.info("【向市局上报数据->从队列取出红狮水泥数据发送】报文dataId：{}",data.getDataId());
                        // 消费数据的逻辑
                        sendReportPackage(data);
                        reportPkgHSSN = data;
                        reSendNumHSSN = new AtomicInteger(3);
                        createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN);
                    }else{
                        Thread.sleep(30_000);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    semaphoreHSSN.release();
                    log.error("【向市局上报数据->从队列取出红狮水泥数据发送】处理中断：{}",e.getMessage());
                }
            }
        }
    }

    private class CYKYDataConsumeThread implements Runnable {
        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    if( reportNettyClient.isConnected() ){
                        // 在收到前一个报文接收成功的响应信息后才可传递下一个报文
                        semaphoreCYKY.acquire();
                        // 从队列中取出数据，如果队列为空则阻塞等待
                        ReportRequestPkgDto data = concurrentHashMap.get(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY).take();
                        log.info("【向市局上报数据->从队列取出昌宇矿业数据发送】报文dataId：{}",data.getDataId());
                        // 消费数据的逻辑
                        sendReportPackage(data);
                        reportPkgCYKY = data;
                        reSendNumCYKY = new AtomicInteger(3);
                        createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY);
                    }else{
                        Thread.sleep(30_000);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    semaphoreCYKY.release();
                    log.error("【向市局上报数据->从队列取出昌宇矿业数据发送】处理中断：{}",e.getMessage());
                }
            }
        }
    }

    /**
     * 通过连接发送报文
     * @param reportRequestPkgDto
     */
    public void sendReportPackage(ReportRequestPkgDto reportRequestPkgDto){
        byte[] req = (JSONObject.toJSONString(reportRequestPkgDto)+CommonConstant.REPORT_DELIMITER).getBytes();
        log.info("【向市局上报数据->通过连接发送报文】报文dataId：{}", reportRequestPkgDto.getDataId());
        reportNettyClient.writeMsg(req);
    }

    @Override
    public void onMessageReceived(String message) {
        if( StringUtils.isEmpty(message) ){
            log.info("【向市局上报数据->接收到响应报文】内容为空！");
            return;
        }
        ReportResponsePkgDto reportResponsePkgDto = JSONObject.parseObject(message, ReportResponsePkgDto.class);
        if( reportResponsePkgDto == null ){
            log.info("【向市局上报数据->接收到响应报文】报文转换失败！");
            return;
        }
        if( reportResponsePkgDto.getSuccess() ){
            if( reportPkgNFSN != null && reportResponsePkgDto.getDataId().equals(reportPkgNFSN.getDataId()) ){
                log.info("【向市局上报数据->南方水泥报文发送】消息dataId：{}上报成功，继续传递下一个报文", reportResponsePkgDto.getDataId());
                cancelTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN);
                semaphoreNFSN.release();
            }else if( reportPkgHSSN != null && reportResponsePkgDto.getDataId().equals(reportPkgHSSN.getDataId()) ){
                log.info("【向市局上报数据->红狮水泥报文发送】消息dataId：{}上报成功，继续传递下一个报文", reportResponsePkgDto.getDataId());
                cancelTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN);
                semaphoreHSSN.release();
            }else if( reportPkgCYKY != null && reportResponsePkgDto.getDataId().equals(reportPkgCYKY.getDataId()) ){
                log.info("【向市局上报数据->昌宇矿业报文发送】消息dataId：{}上报成功，继续传递下一个报文", reportResponsePkgDto.getDataId());
                cancelTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY);
                semaphoreCYKY.release();
            }
        }else{
            if( reportPkgNFSN != null && reportResponsePkgDto.getDataId().equals(reportPkgNFSN.getDataId()) ){
                cancelTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN);
                if( reSendNumNFSN.get() == 0 ){
                    log.info("【向市局上报数据->南方水泥报文发送】消息dataId：{}上报失败：{}，重传次数已达到限定次数，继续传递下一个报文", reportResponsePkgDto.getDataId(), reportResponsePkgDto.getError().getMessage());
                    semaphoreNFSN.release();
                }else{
                    log.info("【向市局上报数据->南方水泥报文发送】消息dataId：{}上报失败：{}，尝试重传", reportResponsePkgDto.getDataId(), reportResponsePkgDto.getError().getMessage());
                    sendReportPackage(reportPkgNFSN);
                    reSendNumNFSN.decrementAndGet();
                    createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN);
                }
            }else if( reportPkgHSSN != null && reportResponsePkgDto.getDataId().equals(reportPkgHSSN.getDataId()) ){
                cancelTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN);
                if( reSendNumHSSN.get() == 0 ){
                    log.info("【向市局上报数据->红狮水泥报文发送】消息dataId：{}上报失败：{}，重传次数已达到限定次数，继续传递下一个报文", reportResponsePkgDto.getDataId(), reportResponsePkgDto.getError().getMessage());
                    semaphoreHSSN.release();
                }else{
                    log.info("【向市局上报数据->红狮水泥报文发送】消息dataId：{}上报失败：{}，尝试重传", reportResponsePkgDto.getDataId(), reportResponsePkgDto.getError().getMessage());
                    sendReportPackage(reportPkgHSSN);
                    reSendNumHSSN.decrementAndGet();
                    createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN);
                }
            }else if( reportPkgCYKY != null && reportResponsePkgDto.getDataId().equals(reportPkgCYKY.getDataId()) ){
                cancelTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY);
                if( reSendNumCYKY.get() == 0 ){
                    log.info("【向市局上报数据->昌宇矿业报文发送】消息dataId：{}上报失败：{}，重传次数已达到限定次数，继续传递下一个报文", reportResponsePkgDto.getDataId(), reportResponsePkgDto.getError().getMessage());
                    semaphoreCYKY.release();
                }else{
                    log.info("【向市局上报数据->昌宇矿业报文发送】消息dataId：{}上报失败：{}，尝试重传", reportResponsePkgDto.getDataId(), reportResponsePkgDto.getError().getMessage());
                    sendReportPackage(reportPkgCYKY);
                    reSendNumCYKY.decrementAndGet();
                    createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY);
                }
            }
        }
    }

    /**
     * 设置超时任务，在30秒后检查是否收到响应，若未收到则释放信号量
     *
     * @param enterpriseName
     */
    private void createTimeoutTask(String enterpriseName) {
        switch (enterpriseName){
            case CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN:
                log.info("【向市局上报数据->南方水泥报文发送】创建响应报文等待任务");
                timeoutTaskNFSN = executorServiceNFSN.schedule(() -> {
                    if (!semaphoreNFSN.tryAcquire()) {
                        if( reSendNumNFSN.get() == 0 ){
                            log.info("【向市局上报数据->南方水泥报文发送】消息dataId：{}上报失败：响应超时，重传次数已达到限定次数，继续传递下一个报文", reportPkgNFSN.getDataId());
                            semaphoreNFSN.release();
                        }else{
                            log.info("【向市局上报数据->南方水泥报文发送】消息dataId：{}上报失败：响应超时，尝试重传", reportPkgNFSN.getDataId());
                            sendReportPackage(reportPkgNFSN);
                            reSendNumNFSN.decrementAndGet();
                            createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN);
                        }
                    }
                }, 30, TimeUnit.SECONDS);
                break;
            case CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN:
                log.info("【向市局上报数据->红狮水泥报文发送】创建响应报文等待任务");
                timeoutTaskHSSN = executorServiceHSSN.schedule(() -> {
                    if (!semaphoreHSSN.tryAcquire()) {
                        if( reSendNumHSSN.get() == 0 ){
                            log.info("【向市局上报数据->红狮水泥报文发送】消息dataId：{}上报失败：响应超时，重传次数已达到限定次数，继续传递下一个报文", reportPkgHSSN.getDataId());
                            semaphoreHSSN.release();
                        }else{
                            log.info("【向市局上报数据->红狮水泥报文发送】消息dataId：{}上报失败：响应超时，尝试重传", reportPkgHSSN.getDataId());
                            sendReportPackage(reportPkgHSSN);
                            reSendNumHSSN.decrementAndGet();
                            createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN);
                        }
                    }
                }, 30, TimeUnit.SECONDS);
                break;
            case CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY:
                log.info("【向市局上报数据->昌宇矿业报文发送】创建响应报文等待任务");
                timeoutTaskCYKY = executorServiceCYKY.schedule(() -> {
                    if (!semaphoreCYKY.tryAcquire()) {
                        if( reSendNumCYKY.get() == 0 ){
                            log.info("【向市局上报数据->昌宇矿业报文发送】消息dataId：{}上报失败：响应超时，重传次数已达到限定次数，继续传递下一个报文", reportPkgCYKY.getDataId());
                            semaphoreCYKY.release();
                        }else{
                            log.info("【向市局上报数据->昌宇矿业报文发送】消息dataId：{}上报失败：响应超时，尝试重传", reportPkgCYKY.getDataId());
                            sendReportPackage(reportPkgCYKY);
                            reSendNumCYKY.decrementAndGet();
                            createTimeoutTask(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY);
                        }
                    }
                }, 30, TimeUnit.SECONDS);
                break;
            default:
                break;
        }
    }

    /**
     * 取消超时任务
     *
     * @param enterpriseName
     */
    private void cancelTimeoutTask(String enterpriseName) {
        switch (enterpriseName){
            case CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN:
                log.info("【向市局上报数据->南方水泥报文发送】收到响应报文，取消等待任务");
                if (timeoutTaskNFSN != null && !timeoutTaskNFSN.isDone()) {
                    timeoutTaskNFSN.cancel(false);
                }
                break;
            case CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN:
                log.info("【向市局上报数据->红狮水泥报文发送】收到响应报文，取消等待任务");
                if (timeoutTaskHSSN != null && !timeoutTaskHSSN.isDone()) {
                    timeoutTaskHSSN.cancel(false);
                }
                break;
            case CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY:
                log.info("【向市局上报数据->昌宇矿业报文发送】收到响应报文，取消等待任务");
                if (timeoutTaskCYKY != null && !timeoutTaskCYKY.isDone()) {
                    timeoutTaskCYKY.cancel(false);
                }
                break;
            default:
                break;
        }
    }

    /**
     * 数据变化情况监测
     */
    @Scheduled(cron = "0 0/10 * * * ?")
    public void monitor(){
        for (Map.Entry<String,LinkedBlockingQueue<ReportRequestPkgDto>> entry : concurrentHashMap.entrySet()) {
            log.info("【向市局上报数据->队列数据存储情况】队列名：{}，当前数据数量：{}",entry.getKey(),entry.getValue().size());
        }
    }
}
